package com.zc.standalone.configuration;

import lombok.AccessLevel;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import com.zc.core.ProtocolSupport;
import com.zc.core.cluster.ClusterManager;
import com.zc.supports.protocol.StaticProtocolSupports;
import com.zc.supports.protocol.management.ProtocolSupportDefinition;
import com.zc.supports.protocol.management.ProtocolSupportLoader;
import com.zc.supports.protocol.management.ProtocolSupportManager;
import org.springframework.boot.CommandLineRunner;
import reactor.core.publisher.Mono;

import java.time.Duration;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Consumer;

/**
 * 懒加载协议支持
 */
@Slf4j
@Getter
@Setter
public class LazyInitManagementProtocolSupports extends StaticProtocolSupports implements CommandLineRunner {

    /*协议管理支持*/
    private ProtocolSupportManager manager;

    /*协议加载支持*/
    private ProtocolSupportLoader loader;

    /*集群管理器*/
    private ClusterManager clusterManager;

    @Setter(AccessLevel.PRIVATE)
    private Map<String, String> configProtocolIdMapping = new ConcurrentHashMap<>();

    // 加载超时时间
    private Duration loadTimeOut = Duration.ofSeconds(30);

    /**
     * 协议初始化
     */
    public void init() {

        clusterManager
                // 获取集群广播
                .<ProtocolSupportDefinition>getTopic("_protocol_changed")
                .subscribe()
                .subscribe(protocol -> this.init(protocol).subscribe());

        try {
            manager.loadAll()
                    .filter(de -> de.getPublishState() == 1)
                    .flatMap(this::init)
                    .blockLast(loadTimeOut);
        } catch (Throwable e) {
            log.error("load protocol error", e);
        }

    }

    /**
     * 协议初始化
     * 1.移除旧协议
     * 2.移除未发布的协议，注册已发布的协议
     * 3.加载已发布的协议
     * @param definition 协议支持参数定义
     * @return
     */
    public Mono<Void> init(ProtocolSupportDefinition definition) {
        // 协议未发布
        if (definition.getPublishState() != 1) {
            // 获取协议id
            String protocol = configProtocolIdMapping.get(definition.getId());
            // 协议id存在
            if (protocol != null) {
                log.debug("uninstall protocol:{}", definition);
                // 卸载协议
                unRegister(protocol);
                return Mono.empty();
            }
        }
        String operation = definition.getPublishState() != 1 ? "uninstall" : "install";

        // 协议已发布则注册协议，否则卸载协议（回调函数：1）
        Consumer<ProtocolSupport> consumer = definition.getPublishState() != 1 ? this::unRegister : this::register;

        log.debug("{} protocol:{}", operation, definition);

        // 加载协议
        return loader
                // 加载协议
                .load(definition)
                .doOnNext(e -> {
                    log.debug("{} protocol[{}] success: {}", operation, definition.getId(), e);
                    configProtocolIdMapping.put(definition.getId(), e.getId());
                    // 回调 1
                    consumer.accept(e);
                })
                .onErrorResume((e) -> {
                    log.error("{} protocol[{}] error: {}", operation, definition.getId(), e);
                    return Mono.empty();
                })
                .then();

    }

    /**
     * 容器启动成功后主动调用
     * @param args
     */
    @Override
    public void run(String... args) {
        init();
    }
}